package com.lz.ad.canal.config.kafka;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by Li Tiefei on 2019/3/9.
 */
@Configuration
@Slf4j
public class KafkaProducerConfig {
    @Autowired
    private KafkaProducerConnectionInfo kafkaConnectionInfo;

    @Bean
    public Map<String,Object> producerConfigs(){
        Map<String, Object> props = new HashMap<>();
        /** 可配置参数(zookeeper) */
        /** 与kafka集群的连接，逗号分隔*/
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConnectionInfo.getBootstrapServers());
        /** 生产记录发送延迟，
         * 如果batch字节未满，不立即发送消息，而是主动等待若干ms，来达到批处理效果
         * 这可以减少请求数量，但会在消息生产速度不够快的情况下带来些许延迟
         * 默认为0（无延迟）
         */
        props.put(ProducerConfig.LINGER_MS_CONFIG, kafkaConnectionInfo.getLingerMs());
        /** 请求最大大小 默认1048576 */
        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,kafkaConnectionInfo.getMaxRequestSize());
        /**
         * 每当多个记录被发送到同一分区时，生产者将尝试将记录一起批量处理为更少的请求，
         * 这有助于提升客户端和服务器上的性能，此配置控制默认批量大小（以字节为单位），默认值为16384
         */
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,kafkaConnectionInfo.getBatchSize());
        /** 生产者可用于缓冲等待发送到服务器的记录的内存总字节数，默认值为33554432 */
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, kafkaConnectionInfo.getBufferMemory());


        /** 可配置参数(application.yml) */
        /** ID在发出请求时传递给服务器，用于服务器端日志记录 */
        if(StringUtils.isNotBlank(kafkaConnectionInfo.getClientId())){
            props.put(ProducerConfig.CLIENT_ID_CONFIG,kafkaConnectionInfo.getClientId());
        }
        /** 分区器类 */
        if(StringUtils.isNotBlank(kafkaConnectionInfo.getPartitionerClass())){
            props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,kafkaConnectionInfo.getPartitionerClass());
        }
        /** 拦截器类列表 */
        if(StringUtils.isNotBlank(kafkaConnectionInfo.getInterceptorClasses())){
            props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,kafkaConnectionInfo.getInterceptorClasses());
        }

        /** 默认参数 */
        /** key的Serializer类，实现类实现了接口org.apache.kafka.common.serialization.Serializer */
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        /** spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer */
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        /**
         * procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下：
         * acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。
         * acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。
         * acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。
         * 可以设置的值为：all, -1, 0, 1
         */
        props.put(ProducerConfig.ACKS_CONFIG,"all");
        /** 幂等性控制，保证每个消息只被写入到stream一次 */
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);

        return props;
    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());
        log.info("KafkaTemplate is created over.");
        return kafkaTemplate;
    }
}
